package com.sxt.disruptor.server;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.springframework.stereotype.Component;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * DisruptorServer，封装了Disruptor服务，提供 start ，stop，注册消费者，提供发布方法等功能
 * 
 * 该类 可以直接使用，并且独立于业务。该类描述了 Disruptor的核心
 * 
 * 
 * @author Administrator
 * @version v1.0.0
 */
@Component
public class DisruptorServer {
	private static final int BUFFER_SIZE = 1024 * 1024;// 环形队列长度，必须是2的N次方

	private Disruptor<Trade> disruptor;

	// 静态内部类的单例模式
	private static class SingletonHolder {
		private static final DisruptorServer INSTANCE = new DisruptorServer();
	}

	// 对外创建
	public static DisruptorServer getInstance() {
		return SingletonHolder.INSTANCE;
	}

	// 对外不能暴露的接口
	private DisruptorServer() {
	}

	public void start() {
		initDisruptor();
	}

	@SuppressWarnings("deprecation")
	private void initDisruptor() {
		// 1.创建一个线程工厂提供线程来触发Consumer的事件处理
		//ExecutorService executor = Executors.newCachedThreadPool(); // 线程池，构建disruptor需要
		ThreadFactory threadFactory = Executors.defaultThreadFactory();
		// 2.创建工厂
		
		// 3. 构建Disruptor
		disruptor = new Disruptor<Trade>(Trade.ENENT_FACTORY, BUFFER_SIZE, threadFactory, ProducerType.MULTI,
				new BlockingWaitStrategy());

		
		//创建消费者数组
        TradeHandler[] consumers = new TradeHandler[5];
        
        for(int i = 0; i<consumers.length;i++) {
        	consumers[i] = new TradeHandler();
        }
		// 4.注册一个消费者，消费者事先注册好
		disruptor.handleEventsWith(consumers);
		
		// 5.开启并发服务
		disruptor.start();
		// 消费者启动了
		// -------------------------------
		// 下面开始生产

		// RingBuffer<Trade> ringBuffer = disruptor.getRingBuffer();//启动轮转，轮转的size
		// 在构建disruptor指定，基于该机制实现的无锁 高并发

	}

	// 生产者
	public TradeProducer getTradeProducer() {
		// 7.创建ringBuffer容器
		RingBuffer<Trade> ringBuffer = this.disruptor.getRingBuffer();
		// 创建一个新的生产者
		return new TradeProducer(ringBuffer);
	}

	public static void main(String[] args) {
		
		DisruptorServer.getInstance().start(); // 初始化服务

		TradeProducer tradeProducer = DisruptorServer.getInstance().getTradeProducer();

		String data = "tradeProducer";

		tradeProducer.onData(data);
	}
}
